package com.yunxin.yundubbo.connection;

import com.alibaba.fastjson2.JSON;
import com.yunxin.yundubbo.codecs.MessageCodecs;
import com.yunxin.yundubbo.mode.Buffer;
import com.yunxin.yundubbo.mode.Message;
import com.yunxin.yundubbo.util.ChannelCacheUtil;
import com.yunxin.yundubbo.util.ChannelIdContext;
import com.yunxin.yundubbo.util.RequestContext;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class DubboClient {

    private static final Logger logger = LoggerFactory.getLogger(DubboClient.class);

    private static final int MAX_RETRY = 10;
    private static final long INITIAL_RETRY_INTERVAL = 2000; // 2秒初始重试间隔
    private static final String HEART_BEAT = "heart";
    private final NioEventLoopGroup workerGroup = new NioEventLoopGroup();
    private Bootstrap bootstrap;

    private static volatile DubboClient instance;

    private DubboClient() {
        initBootstrap();
    }

    public static DubboClient getInstance() {
        if (instance == null) {
            synchronized (DubboClient.class) {
                if (instance == null) {
                    instance = new DubboClient();
                }
            }
        }
        return instance;
    }


    /**
     * 初始化Bootstrap和Pipeline
     */
    private void initBootstrap() {
        bootstrap = new Bootstrap()
                .group(workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(102400,6 , 4, 0, 0));
                        // 30秒无写操作，触发心跳
                        pipeline.addLast(new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS));

                        // 自定义消息编解码器
                        pipeline.addLast(new MessageCodecs());

                        // 消息处理
                        pipeline.addLast(new MessageHandler());

                        // 心跳处理
                        pipeline.addLast(new HeartbeatHandler());
                    }
                });
    }

    /**
     * 第一次连接
     */
    public void connect(String host, int port, String serverName) {
        doConnect(0, host, port, serverName);
    }


    /**
     * 安全重连
     */
    private void doConnect(int retry, String host, int port, String serverName) {
        if (retry > MAX_RETRY) {
            logger.error("超过最大重连次数，放弃连接！");
            return;
        }

        long delay = Math.min(INITIAL_RETRY_INTERVAL << retry, 30000); // 指数退避，最大30秒
        logger.info("尝试连接 {}:{}，第 {} 次，延迟 {}ms", host, port, retry + 1, delay);

        workerGroup.schedule(() -> bootstrap.connect(host, port).addListener((ChannelFutureListener) future -> {
            System.out.println(future.isSuccess());
            if (future.isSuccess()) {
                logger.info("连接服务器成功！");
                Channel channel = future.channel();
                ChannelIdContext.set(channel.id().asLongText(), serverName);
                ChannelCacheUtil.setCacheHandler(serverName, channel);
            } else {
                logger.warn("连接失败，准备重试...");
                doConnect(retry + 1, host, port, serverName);
            }
        }), delay, TimeUnit.MILLISECONDS);
    }


    public Channel doConnectSync(String host, int port, String serverName) {
        int retry = 0;
        while (retry <= MAX_RETRY) {
            try {
                logger.info("尝试连接 {}:{}，第 {} 次", host, port, retry + 1);
                ChannelFuture future = bootstrap.connect(host, port).sync(); // 阻塞直到连接完成
                if (future.isSuccess()) {
                    logger.info("连接服务器成功！");
                    Channel channel = future.channel();
                    ChannelIdContext.set(channel.id().asLongText(), serverName);
                    ChannelCacheUtil.setCacheHandler(serverName, channel);
                    return channel;
                }
            } catch (Exception e) {
                retry++;
                long delay = Math.min(INITIAL_RETRY_INTERVAL << retry, 30000);
                logger.warn("连接失败，第 {} 次重试，等待 {}ms", retry, delay);
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
        logger.error("超过最大重连次数，放弃连接！");
        return null;
    }


    /**
     * 关闭客户端
     */
    public void shutdown() {
        Channel ch = ChannelCacheUtil.getCacheHandler("default");
        if (ch != null) {
            ch.close();
        }
        workerGroup.shutdownGracefully();
    }

    /**
     * -------------------- 内部Handler --------------------
     */
    //开启handler共享是为了保证多个Channel可以进入其中
    @ChannelHandler.Sharable
    private class MessageHandler extends ChannelInboundHandlerAdapter {

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            Message message = (Message) msg;
            Buffer buffer = JSON.parseObject(message.getBytes(), Buffer.class);
            logger.info("receive a response from the server ，requestId：{}",buffer.getRequestId());
            CompletableFuture<String> future = RequestContext.remove(buffer.getRequestId());
            if (future != null) {
                future.complete(buffer.getData());
            }
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            logger.warn("连接断开，尝试重连...");
            InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
            String serverName = ChannelIdContext.remove(ctx.channel().id().asLongText());
            doConnect(0, inetSocketAddress.getHostString(), inetSocketAddress.getPort(), serverName);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            logger.error("Netty异常: {}", cause.getMessage(), cause);
            ctx.close();
        }
    }

    @ChannelHandler.Sharable
    private static class HeartbeatHandler extends ChannelInboundHandlerAdapter {

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof io.netty.handler.timeout.IdleStateEvent event &&
                    event.state() == io.netty.handler.timeout.IdleState.WRITER_IDLE) {
                // 发送心跳
                logger.info("发送心跳包...");
                ctx.writeAndFlush(HEART_BEAT);
            } else {
                super.userEventTriggered(ctx, evt);
            }
        }
    }
}

